package com.lzwk.app.dwd.ocpx;


import com.lzwk.common.ocpx.OCPXConstant;
import org.apache.tools.ant.taskdefs.Sleep;

import java.util.HashMap;
import java.util.Map;

/**
 * @title: DwdLookUpJoinTables
 * @Author Yeaser
 * @Date: 2022/2/15 14:39
 * @Version 1.0
 * Description: TODO
 */
public class DwdLookUpJoinTablesN {
    public static Map<String, String> dwdPayOrder(){

        //ods pay order
        String sourceTableOrder = "create table mysql_pay_order(" +
                " `id` INT, " +
                " `payee_id` INT, " +
                " `payee_type` STRING, " +
                " `fee` INT, " +
                " `product_type` INT, " +
                " `product_id` INT, " +
                " `source_id` INT, " +
                " `skin_id` INT, " +
                " `wxapp_id` INT, " +
                " `paid` INT, " +
                " `deliver_status` INT, " +
                " `refund_status` INT, " +
                " `create_time` TIMESTAMP(0), " +
                " `pay_time` TIMESTAMP(0), " +
                " `callback_time` TIMESTAMP(0), " +
                " `outer_order_id` STRING, " +
                " `meta` STRING NOT NULL, " +
                " `ip` STRING, " +
                " `payer_id` INT, " +
                " `source` STRING, " +
                " `is_notified` INT, " +
                " `update_time` TIMESTAMP(0), " +
                " `openid` STRING, " +
                " `proc_time` AS PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='" + OCPXConstant.OCPXSLAVEHOST + "', " +
                "   'port'='" + OCPXConstant.OCPXSLAVEPORT + "', " +
                "   'username'='" + OCPXConstant.OCPXSLAVEUSERNAME + "', " +
                "   'password'='" + OCPXConstant.OCPXSLAVEPASSWORD + "', " +
                "   'database-name'='" + OCPXConstant.OCPXSLAVEDBNAME + "', " +
                "   'table-name'='" + OCPXConstant.OCPXJDBCPAYORDER + "', " +
                "   'server-id'='35231-35240', " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none', " +
                "   'scan.incremental.snapshot.chunk.size' = '80000' " +
                ")";

        String sinkTableOrder ="create table dwd_ocpx_pay_order(" +
                "   `id` INT, " +
                "   `payer_id` INT, " +
                "   `product_id` INT, " +
                "   `wk_account_id` INT, " +
                "   `parent_activity_id` INT, " +
                "   `src_prod_id` INT, " +
                "   `org_id` INT, " +
                "   `ks_aid` STRING, " +
                "   `product_name` STRING, " +
                "   `skin_id` INT, " +
                "   `skin_name` STRING, " +
                "   `source_id` INT, " +
                "   `source_name` STRING, " +
                "   `wxapp_id` INT," +
                "   `wxapp_nickname` STRING," +
                "   `nickname` STRING," +
                "   `headimgurl` STRING," +
                "   `phone_num` STRING, " +
                "   `creative_id` STRING, " +
                "   `ad_plan_id` STRING, " +
                "   `ad_group_id` STRING, " +
                "   `gdt_group_id` STRING, " +
                "   `bd_creative_id` STRING, " +
                "   `fee` INT, " +
                "   `paid` INT, " +
                "   `create_time` STRING, " +
                "   `pay_time` STRING, " +
                "   `subscribed` INT, " +
                "   `tag_id` INT, " +
                "   `tag_name` STRING, " +
                "   `source_platform` STRING, " +
                "   `refund_status` INT, " +
                "   `meta` STRING, " +
                "   `hash_id` STRING, " +
                "   `trace_id` STRING, " +
                "   `appid` STRING, " +
                "   PRIMARY KEY (`id`) NOT ENFORCED" +
                ")with(" +
                "   'connector'='doris', " +
                "   'fenodes'='"+ OCPXConstant.OCPXDORISIP +"', " +
                "   'table.identifier'='"+ OCPXConstant.OCPXDORISDATABASE + "." + OCPXConstant.OCPXDWDPAYORDER +"', " +
                "   'username'='"+ OCPXConstant.OCPXDORISUSERNAME +"'," +
                "   'password'='"+ OCPXConstant.OCPXDORISPASSWORD +"'," +
                "   'sink.batch.size'='"+ OCPXConstant.OCPXDORISSINKBATCHSIZE +"', " +
                "   'sink.batch.interval'='"+ OCPXConstant.OCPXDORISSINKBATCHINTERVAL +"' " +
                ")";

        //dwd
        String insertTableOrder = "  insert into  dwd_ocpx_pay_order " +
                " select " +
                " ifnull(id,-1)  as id, " +
                " ifnull(payer_id,-1)  as payer_id, " +
                " ifnull(product_id,-1) as product_id, " +
                " ifnull(wk_account_id,-1) as wk_account_id, " +
                " ifnull(parent_activity_id,-1) as parent_activity_id, " +
                " ifnull(src_prod_id,-1) as src_prod_id, " +
                " ifnull(org_id,-1) as org_id,  " +
                " ifnull(ks_aid,'') as ks_aid, " +
                " ifnull(product_name,'') as product_name, " +
                " ifnull(skin_id,-1) as skin_id, " +
                " ifnull(skin_name,'') as skin_name, " +
                " ifnull(source_id,-1)  as source_id, " +
                " ifnull(source_name,'') as source_name, " +
                " ifnull(wxapp_id,-1) as wxapp_id, " +
                " ifnull(wxapp_nickname,'') as wxapp_nickname,  " +
                " ifnull(nickname,'') as nickname, " +
                " ifnull(headimgurl,'') as headimgurl, " +
                " ifnull(phone_num,'') as phone_num, " +
                " ifnull(creative_id,'') as creative_id, " +
                " ifnull(ad_plan_id,'') as ad_plan_id, " +
                " ifnull(ad_group_id,'') as ad_group_id, " + // MP消耗与订单关联必-1) 须做限制
                " ifnull( gdt_group_id,'') as gdt_group_id, " + //--广点通消耗表与订-1) 单表关联
                " ifnull( bd_creative_id,'') as bd_creative_id, " + //--广点通消耗表与订-1) 单表关联
                " ifnull(fee,0) as fee, " +
                " ifnull(paid,0) as paid, " +
                " ifnull(create_time,'') as create_time, " +
                " ifnull(pay_time,'') as pay_time, " +
                " subscribed, " +
                " ifnull(tag_id,100) as tag_id, " +
                " ifnull(tag_name ,'') as tag_name, " +
                " ifnull(source_platform,'')  as source_platform, " +
                " ifnull(refund_status,0) as refund_status, " +
                " ifnull(meta,'') as meta, " +
                " ifnull(hash_id,'') as hash_id, " +
                " ifnull(trace_id,'') as trace_id, " +
                " ifnull(appid,'') as appid " +
                " from (select  " +
                        " id, " +
                        " payer_id, " +
                        " product_id, " +
                        " wk_account_id, " +
                        " parent_activity_id, " +
                        " src_prod_id, " +
                        " org_id, " +
                        " ks_aid, " +
                        " product_name, " +
                        " skin_id, " +
                        " skin_name, " +
                        " source_id, " +
                        " source_name, " +
                        " wxapp_id, " +
                        " wxapp_nickname, " +
                        " nickname, " +
                        " headimgurl, " +
                        " phone_num, " +
                        " creative_id, " +
                        " ad_plan_id, " +
                        " if(`method`='wx_jsapi', if(wx_aid <> '',wx_aid, ad_group_id),'') AS ad_group_id, " + // MP消耗与订单关联必须做限制
                        " ad_group_id AS gdt_group_id, " + //--广点通消耗表与订单表关联
                        " bd_creative_id, " +
                        " fee, " +
                        " paid, " +
                        " create_time, " +
                        " pay_time, " +
                        " subscribed, " +
                        " tag_id, " +
                        " tag_name , " +
                        " source_platform, " +
                        " refund_status, " +
                        " meta, " +
                        " hash_id, " +
                        " trace_id, " +
                        " appid " +
                        " from ( " +
                        "   select " +
                        "   t1.payer_id, " +
                        "   t1.product_id," +
                        "   ifnull(t5.account_id,-1) as wk_account_id, " +
                        "   ifnull(t8.parent_id,-1) as parent_activity_id, " +
                        "   t1.src_prod_id, " +
                        "   t1.org_id, " +
                        "   t1.ks_aid, " +
                        "   t1.bd_creative_id, " +
                        "   t1.product_name, " +
                        "   t1.skin_id, " +
                        "   t1.skin_name, " +
                        "   t1.source_id, " +
                        "   t1.source_name, " +
                        "   t1.wxapp_id, " +
                        "   t1.wxapp_nickname, " +
                        "   ifnull(if(t2.src='wx',regexp_replace(t2.nickname,'\\n|\\t',''),regexp_replace(t6.nickname,'\\n|\\t','')),'') as nickname, " +
                        "   ifnull(if(t2.src='wx',regexp_replace(t2.headimgurl,'\\n|\\t',''),regexp_replace(t6.headimgurl,'\\n|\\t','')),'') as headimgurl," +
                        "   ifnull(if(t2.src<>'wx', regexp_replace(t2.openid,'o_',''), regexp_replace(t7.openid,'o_','')),'') as phone_num, " +
                        "   t1.creative_id," +
                        "   GetJsonString(t1.meta,'ad_plan_id') as ad_plan_id," +
                        "   ifnull(if(starts_with(source_platform,'gdt'),t1.gdt_id,regexp_replace(`t4`.`ad_group_id`,'\\n|\\t','')),'') as ad_group_id, " +
                        "   t1.fee, " +
                        "   t1.paid, " +
                        "   CAST(DATE_FORMAT(t1.create_time,'yyyy-MM-dd HH:mm:ss') AS STRING) as create_time, " +
                        "   CAST(DATE_FORMAT(t1.pay_time,'yyyy-MM-dd HH:mm:ss') AS STRING) as pay_time, " +
                        "   t1.wx_aid, " +
                        "   t1.`method`, " +
                        "   t3.subscribed, " +
                        "   t1.id, " +
                        "   t1.tag_id, " +
                        "   t1.tag_name, " +
                        "   t1.source_platform, " +
                        "   t1.refund_status, " +
                        "   ifnull(t1.hash_id,'') as hash_id, " +
                        "   ifnull(t4.trace_id,'') as trace_id, " +
                        "   ifnull(t9.appid,'') as appid, " +
                        "   t1.meta " +
//                        "   row_number() over (partition by t1.id order by t4.subscribe_time desc) AS rn " +
                        "   from " +
                        "       (select " +
                        "            tt1.`payer_id`, " +
                        "            tt1.`product_id`, " +
                        "            ifnull(tt1.`source_id`, 1) AS `source_id`, " +
                        "            ifnull(tt1.`skin_id`, -1) AS `skin_id`, " +
                        "            ifnull(tt1.`wxapp_id`, -1) AS `wxapp_id`, " +
                        "            ifnull(tt7.`org_id`, -1) AS `org_id`, " +
                        "            ifnull(tt5.`tag_id`, -1) AS `tag_id`, " +
                        "            ifnull(regexp_replace(tt3.`name`,'\\n|\\t',''), '') AS `source_name`, " +
                        "            ifnull(regexp_replace(tt4.`name`,'\\n|\\t',''), '') AS `skin_name`, " +
                        "            ifnull(regexp_replace(tt5.`name`,'\\n|\\t',''), '') AS `product_name`, " +
                        "            ifnull(regexp_replace(tt8.`tag_name`,'\\n|\\t',''), '') AS `tag_name`, " +
                        "            ifnull(regexp_replace(tt6.`nickname`,'\\n|\\t',''), '') AS `wxapp_nickname`, " +
                        "            ifnull(tt5.`src_prod_id`,-1) as src_prod_id, " +
                        "            ifnull(CAST(REGEXP_EXTRACT(GetJsonString(tt1.`meta`, 'url'), 'creativeid=([0-9]+)', 1) AS STRING),'') AS creative_id, " +
                        "            ifnull(CAST(REGEXP_EXTRACT(GetJsonString(tt1.`meta`, 'url'), 'wx_aid=([0-9]+)', 1) AS STRING),'') AS wx_aid, " +
                        "            ifnull(CAST(REGEXP_EXTRACT(GetJsonString(tt1.`meta`, 'url'), 'ks_aid=([0-9]+)', 1) AS STRING)  ,'') AS ks_aid, " +
                        "            ifnull(CAST(REGEXP_EXTRACT(GetJsonString(tt1.`meta`, 'url'), 'creative=([0-9]+)', 1) AS STRING)  ,'') AS bd_creative_id, " +
                        "            ifnull(GetJsonString(tt1.`meta`, 'openid'),'') AS `open_id`, " +
                        "            if(tt1.source='dyshop',GetJsonInt(tt1.`meta`,'pay_amount',0),tt1.`fee`) AS `fee`, " +
                        "            ifnull(GetJsonString(tt1.`meta`, 'method'),'') AS `method`, " +
                        "            tt1.`payee_id`, " +
                        "            ifnull(regexp_replace(tt1.`payee_type`,'\\n|\\t',''),'') AS `payee_type`, " +
                        "            ifnull(tt1.`product_type`,-1) AS product_type, " +
                        "            tt1.`paid`, " +
                        "            tt1.`deliver_status`, " +
                        "            tt1.`refund_status`, " +
                        "            ifnull(tt1.`create_time`,TO_TIMESTAMP('1970-01-01 08:00:01')) AS `create_time`, " +
                        "            ifnull(tt1.`pay_time`,TO_TIMESTAMP('1970-01-01 08:00:01')) AS `pay_time`, " +
                        "            ifnull(tt1.`callback_time`,TO_TIMESTAMP('1970-01-01 08:00:01'))  AS `callback_time`, " +
                        "            ifnull(regexp_replace(tt1.`outer_order_id`,'\\n|\\t',''),'') AS `outer_order_id`, " +
                        "            ifnull(regexp_replace(tt1.`ip`,'\\n|\\t',''),'') AS `ip`, " +
                        "            ifnull(regexp_replace(tt1.`source`,'\\n|\\t',''),'') AS `source`, " +
                        "            ifnull(tt1.`is_notified`,-1) AS `is_notified`, " +
                        "            ifnull(tt1.`update_time`, TO_TIMESTAMP('1970-01-01 08:00:01')) AS `update_time`, " +
                        "            ifnull(regexp_replace(tt1.`openid`,'\\n|\\t',''),'') AS `openid`, " +
                        "            ifnull(regexp_replace(tt1.`meta`,'\\n|\\t',''),'') AS `meta`, " +
                        "            tt1.`id`, " +
                        "            ifnull(tt3.`meta`,'') AS `platform_meta`, " +
                        "            ifnull(GetJsonString(tt3.`meta`,'source_platform'),'') AS `source_platform`, " +
                        "            ifnull(tt3.hash_id,'') AS `hash_id`, " +
                        "            tt1.`proc_time`, " +
                        "            case when starts_with(ifnull(GetJsonString(tt3.`meta`,'source_platform'),''),'gdt') " +
                        "                 then if(ifnull(regexp_extract(GetJsonString(ifnull(regexp_replace(tt1.`meta`,'\\n|\\t',''),''),'url'),'wx_aid=([0-9]+)',1),'')='', " +
                        "                     regexp_extract(GetJsonString(ifnull(regexp_replace(tt1.`meta`,'\\n|\\t',''),''),'url'),'adgroupid=([0-9]+)',1), " +
                        "                     regexp_extract(GetJsonString(ifnull(regexp_replace(tt1.`meta`,'\\n|\\t',''),''),'url'),'wx_aid=([0-9]+)',1)) " +
                        "                else '' " +
                        "            end as gdt_id " +
                        "            FROM mysql_pay_order   tt1 " +
                        "            left join ocpx_promote_promotesource FOR SYSTEM_TIME AS OF `tt1`.`proc_time` as tt3 " +
                        "            on tt1.source_id=tt3.id " +
                        "            left join ocpx_product_productskin FOR SYSTEM_TIME AS OF `tt1`.`proc_time` as tt4 " +
                        "            on tt1.skin_id=tt4.id " +
                        "            left join ocpx_product_product FOR SYSTEM_TIME AS OF `tt1`.`proc_time` as tt5 " +
                        "            on tt1.product_id = tt5.id " +
                        "            left join ocpx_wxapp_wxapp FOR SYSTEM_TIME AS OF `tt1`.`proc_time` tt6 " +
                        "            on tt1.wxapp_id = tt6.id " +
                        "            left join ocpx_public_admin_recordorgref FOR SYSTEM_TIME AS OF `tt1`.`proc_time` as tt7 " +
                        "            on tt1.product_id=tt7.object_id " +
                        "            and tt7.object_type = 'product' " +
                        "            left join ocpx_tag_info FOR SYSTEM_TIME AS OF `tt1`.`proc_time` as tt8 " +
                        "            on tt5.tag_id=tt8.id " +
                        "            where tt5.src_prod_type = 'wk_act'   "+
                        " ) as t1 " +
                        "   left join ods_ocpx_account_account  FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t2 " +
                        "       on t1.payer_id = t2.id " +
                        "   left join ods_ocpx_account_wxuser FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t3 " +
                        "       on ifnull(`t2`.`parent_account_id`, `t2`.`id`) = t3.account_id " +
                        "       and t1.wxapp_id = t3.wxapp_id " +
                        "   left join ods_ocpx_wxapp_mpadsubfeedbackdata FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t4 " +
                        "       ON t1.open_id = t4.openid" +
                        "       AND CAST(UNIX_TIMESTAMP(DATE_FORMAT(t1.pay_time,'yyyy-MM-dd HH:mm:ss')) AS STRING) > t4.subscribe_time  " +
                        "   left join adb_account_mapping FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t5 " +
                        "       on CAST(t1.payer_id AS STRING) = t5.object_id  " +
                        "       and t5.object_type = 'ocpx' " +
                        "   left join ods_ocpx_account_account FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t6 " + //得到父id的昵称和头像(wx进来直接取,非wx进来的需要取到绑定的父id的微信信息)
                        "       on t2.parent_account_id = t6.id " +
                        "   left join ods_ocpx_account_account FOR SYSTEM_TIME AS OF `t1`.`proc_time` as  t7 " + //得到电话号码(非wx进来的需要取到他)
                        "       on t2.id = t7.parent_account_id " +
                        "   left join adb_division_activity FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t8 " +
                        "       on t1.src_prod_id=t8.id " +
                        "   left join ocpx_wxapp_wxapp FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t9 " +
                        "       on regexp_replace(`t4`.`wechat_account_id`,'\\n|\\t','') = `t9`.`appid` " +
                        "       ) t ) m  ";

        Map<String, String> res = new HashMap<>();
        res.put("source",sourceTableOrder);
        res.put("sink",sinkTableOrder);
        res.put("insert",insertTableOrder);

        return res;

/*        tenv
                .toRetractStream(intervalTable, com.lzwk.beans.ocpx.DwdLookUpJoinTables.class)
                .filter(data -> data.f0)
                .map(data -> data.f1)
                .addSink(MyKafkaUtil.getKafkaProducer(OCPXConstant.OCPXTESTTOPIC2, ""));*/


        /* 启动命令
        *  flink run -n -t yarn-per-job  -Dyarn.application.name=ocpx_kafka_test_v1 -Djobmanager.memory.process.size=2048MB -Dtaskmanar.memory.process.size=8192mb -Dtaskmanager.memory.managed.size=4096MB -Dtaskmanager.numberOfTaskSlots=4 -Dstate.backend.incremental=true -Dstate.backend.local-recovery=true -Dstate.backend.rocksdb.predefined-options=SPINNING_DISK_OPTIMIZED_HIGH_MEM -Dstate.backend.rocksdb.block.cache-size=64MB -Dstate.backend.rocksdb.writebuffer.size=128MB -Dstate.backend.rocksdb.compaction.level.max-size-level-base=320MB -Dstate.backend.rocksdb.writebuffer.count=5 -Dstate.backend.rocksdb.thread.num=4 -Dstate.backend.rocksdb.writebuffer.number-to-merge=3 -Dstate.backend.rocksdb.memory.partitioned-index-filters=true -Dstate.backend.latency-track.keyed-state-enabled=true -c com.lzwk.app.OcpxSum  /home/hadoop/lc/myopt/module/flink/jar/Test/TestProV1.jar
        *
        *
        * */


    }

    public static Map<String, String> odsWxuser(){
        String sourceTable = "create table ocpx_account_wxuser(" +
                " `id` INT, " +
                " `openid` STRING, " +
                " `unionid` STRING, " +
                " `headimgurl` STRING, " +
                " `sex` INT, " +
                " `nickname` STRING, " +
                " `subscribed` INT, " +
                " `create_time` TIMESTAMP(0), " +
                " `active_ip` STRING, " +
                " `active_time` TIMESTAMP(0), " +
                " `update_time` TIMESTAMP(0), " +
                " `account_id` INT, " +
                " `wxapp_id` INT, " +
                "  `proc_time` AS PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='" + OCPXConstant.OCPXSLAVEHOST + "', " +
                "   'port'='" + OCPXConstant.OCPXSLAVEPORT + "', " +
                "   'username'='" + OCPXConstant.OCPXSLAVEUSERNAME + "', " +
                "   'password'='" + OCPXConstant.OCPXSLAVEPASSWORD + "', " +
                "   'database-name'='" + OCPXConstant.OCPXSLAVEDBNAME + "', " +
                "   'table-name'='" + OCPXConstant.OCPXJDBCACCOUNTWXUSER + "', " +
                "   'server-id'='35241-35250', " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none', " +
                "   'scan.incremental.snapshot.chunk.size' = '80000' " +
                ")";


        String sinkTable ="create table dwd_ocpx_pay_order1(" +
                "   `id` INT, " +
                "   `payer_id` INT, " +
                "   `product_id` INT, " +
                "   `wk_account_id` INT, " +
                "   `parent_activity_id` INT, " +
                "   `src_prod_id` INT, " +
                "   `org_id` INT, " +
                "   `ks_aid` STRING, " +
                "   `product_name` STRING, " +
                "   `skin_id` INT, " +
                "   `skin_name` STRING, " +
                "   `source_id` INT, " +
                "   `source_name` STRING, " +
                "   `wxapp_id` INT," +
                "   `wxapp_nickname` STRING," +
                "   `nickname` STRING," +
                "   `headimgurl` STRING," +
                "   `phone_num` STRING, " +
                "   `creative_id` STRING, " +
                "   `ad_plan_id` STRING, " +
                "   `ad_group_id` STRING, " +
                "   `gdt_group_id` STRING, " +
                "   `bd_creative_id` STRING, " +
                "   `fee` INT, " +
                "   `paid` INT, " +
                "   `create_time` STRING, " +
                "   `pay_time` STRING, " +
                "   `subscribed` INT, " +
                "   `tag_id` INT, " +
                "   `tag_name` STRING, " +
                "   `source_platform` STRING, " +
                "   `refund_status` INT, " +
                "   `meta` STRING, " +
                "   `hash_id` STRING, " +
                "   `trace_id` STRING, " +
                "   `appid` STRING, " +
                "   PRIMARY KEY (`id`) NOT ENFORCED" +
                ")with(" +
                "   'connector'='doris', " +
                "   'fenodes'='"+ OCPXConstant.OCPXDORISIP +"', " +
                "   'table.identifier'='"+ OCPXConstant.OCPXDORISDATABASE + "." + OCPXConstant.OCPXDWDPAYORDER +"', " +
                "   'username'='"+ OCPXConstant.OCPXDORISUSERNAME +"'," +
                "   'password'='"+ OCPXConstant.OCPXDORISPASSWORD +"'," +
                "   'sink.batch.size'='"+ OCPXConstant.OCPXDORISSINKBATCHSIZE +"', " +
                "   'sink.batch.interval'='"+ OCPXConstant.OCPXDORISSINKBATCHINTERVAL +"' " +
                ")";


        String insertTable = " insert into  dwd_ocpx_pay_order1 " +
                " select  " +
                " id, " +
                " payer_id, " +
                " product_id, " +
                " wk_account_id, " +
                " parent_activity_id, " +
                " src_prod_id, " +
                " org_id, " +
                " ks_aid, " +
                " product_name, " +
                " skin_id, " +
                " skin_name, " +
                " source_id, " +
                " source_name, " +
                " wxapp_id, " +
                " wxapp_nickname, " +
                " nickname, " +
                " headimgurl, " +
                " phone_num, " +
                " creative_id, " +
                " ad_plan_id, " +
                " ad_group_id, " + // MP消耗与订单关联必须做限制
                " gdt_group_id, " + //--广点通消耗表与订单表关联
                " bd_creative_id, " +
                " fee, " +
                " paid, " +
                " create_time, " +
                " pay_time, " +
                " subscribed, " +
                " tag_id, " +
                " tag_name , " +
                " source_platform, " +
                " refund_status, " +
                " meta, " +
                " hash_id, " +
                " trace_id, " +
                " appid " +
                " from( " +
                " select" +
                " t.id,  " +
                " cast(null as int) as payer_id, " +
                " cast(null as int) as product_id, " +
                " cast(null as int) as wk_account_id, " +
                " cast(null as int) as parent_activity_id, " +
                " cast(null as int) as src_prod_id, " +
                " cast(null as int) as org_id, " +
                " cast(null as varchar) as ks_aid, " +
                " cast(null as varchar) as product_name, " +
                " cast(null as int) as skin_id, " +
                " cast(null as varchar) as skin_name, " +
                " cast(null as int) as source_id, " +
                " cast(null as varchar) as source_name, " +
                " cast(null as int) as wxapp_id, " +
                " cast(null as varchar) as wxapp_nickname, " +
                " cast(null as varchar) as nickname, " +
                " cast(null as varchar) as headimgurl, " +
                " cast(null as varchar) as phone_num, " +
                " cast(null as varchar) as creative_id, " +
                " cast(null as varchar) as ad_plan_id, " +
                " cast(null as varchar) as ad_group_id, " + // MP消耗与订单关联必须做限制
                " cast(null as varchar) as gdt_group_id, " + //--广点通消耗表与订单表关联
                " cast(null as varchar) as bd_creative_id, " +
                " cast(null as int) as fee, " +
                " cast(null as int) as paid, " +
                " cast(null as varchar) as create_time, " +
                " cast(null as varchar) as pay_time, " +
                " t.subscribed as subscribed, " +
                " cast(null as int) as tag_id, " +
                " cast(null as varchar) as tag_name , " +
                " cast(null as varchar) as source_platform, " +
                " cast(null as int) as refund_status, " +
                " cast(null as varchar) as meta, " +
                " cast(null as varchar) as hash_id, " +
                " cast(null as varchar) as trace_id, " +
                " cast(null as varchar) as appid " +
                " from ( " +
                " select  " +
                " order_id as id,  " +
                " t1.subscribed as subscribed " +
                " from ocpx_account_wxuser t1 " +
                " left join ods_ocpx_account_account FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t2 " +
                "   on t1.account_id = t2.parent_account_id  " +
                " join ocpx_pay_order FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t3 " +
                "   on cast(t1.wxapp_id as bigint) = t3.wxapp_id  " +
                "   and t2.id = t3.payer_id " +
                " left join ocpx_pay_order FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t4 " +
                "   on cast(t1.wxapp_id as bigint) = t4.wxapp_id  " +
                "   and t1.account_id = t4.payer_id " +
                " left join ocpx_product_product FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t5 " +
                /*此处有坑
                 *      由于前面关联是或者的关系，因此在此处就需要两次判断
                 * */
                "   on cast(ifnull(t3.product_id,t4.product_id) as int)= t5.id " +
                /*因为是wxuser反查order表拿order_id
                 * 有两种情况：
                 *1、    即用wxuser中的account_id 与account_account中的parent_account_id
                 *       关联拿到account——account中的id，再用此id预订单表中的pay_order中的payer_id关联得到订单id
                 *2、    用wxuser中的account_id直接与pay_order中的payer_id关联得到订单id
                 *       比如wxuser中account_id 为7973024时 以上两种方式都可以关联上，即这一条wxuser的数据对应着pay_order中两条数据；
                 *       一条数据过系统分成两条数据，需用到UDTF
                 * */
                " join lateral table(GetMixedOrderId(if(cast(t3.id as string) is null,if(cast(t4.id as string) is null,'',cast(t4.id as string)),if(cast(t4.id as string) is null,cast(t3.id as string),CONCAT_WS(',',cast(t3.id as string),cast(t4.id as string)))))) as t (order_id) on true " +
                " where t5.src_prod_type = 'wk_act' " +
                ")t)m ";



        Map<String, String> res = new HashMap<>();
        res.put("source",sourceTable);
        res.put("sink",sinkTable);  
        res.put("insert",insertTable);



        return res;
    }

    /* 这里多一条account_account流是因为存在一种情况：
    * account_account中的parent_account_id是会从无到有update，因此parent_account_id作为中间关联维度条件，如果为null那么order流和wxuser流就会关联不上，
    * 因此需要一条account_account流监听parent_account_id的变化
    * */
    public static Map<String, String> odsAccount(){
        String sourceTable = "create table ocpx_account_account(" +
                " `id` INT, " +
                " `openid` STRING, " +
                " `unionid` STRING, " +
                " `nickname` STRING, " +
                " `headimgurl` STRING, " +
                " `sex` INT, " +
                " `province` STRING, " +
                " `country` STRING, " +
                " `city` STRING, " +
                " `status` INT, " +
                " `is_robot` INT, " +
                " `active_ip` STRING, " +
                " `active_time` TIMESTAMP(0), " +
                " `create_time` TIMESTAMP(0), " +
                " `is_init` INT, " +
                " `src` STRING, " +
                " `parent_account_id` INT, " +
                "  `proc_time` AS PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='" + OCPXConstant.OCPXSLAVEHOST + "', " +
                "   'port'='" + OCPXConstant.OCPXSLAVEPORT + "', " +
                "   'username'='" + OCPXConstant.OCPXSLAVEUSERNAME + "', " +
                "   'password'='" + OCPXConstant.OCPXSLAVEPASSWORD + "', " +
                "   'database-name'='" + OCPXConstant.OCPXSLAVEDBNAME + "', " +
                "   'table-name'='" + OCPXConstant.OCPXJDBCACCOUNTACCOUNT + "', " +
                "   'server-id'='35221-35230', " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none', " +
                "   'scan.incremental.snapshot.chunk.size' = '80000' " +
                ")";


        String sinkTable ="create table dwd_ocpx_pay_order2(" +
                "   `id` INT, " +
                "   `payer_id` INT, " +
                "   `product_id` INT, " +
                "   `wk_account_id` INT, " +
                "   `parent_activity_id` INT, " +
                "   `src_prod_id` INT, " +
                "   `org_id` INT, " +
                "   `ks_aid` STRING, " +
                "   `product_name` STRING, " +
                "   `skin_id` INT, " +
                "   `skin_name` STRING, " +
                "   `source_id` INT, " +
                "   `source_name` STRING, " +
                "   `wxapp_id` INT," +
                "   `wxapp_nickname` STRING," +
                "   `nickname` STRING," +
                "   `headimgurl` STRING," +
                "   `phone_num` STRING, " +
                "   `creative_id` STRING, " +
                "   `ad_plan_id` STRING, " +
                "   `ad_group_id` STRING, " +
                "   `gdt_group_id` STRING, " +
                "   `bd_creative_id` STRING, " +
                "   `fee` INT, " +
                "   `paid` INT, " +
                "   `create_time` STRING, " +
                "   `pay_time` STRING, " +
                "   `subscribed` INT, " +
                "   `tag_id` INT, " +
                "   `tag_name` STRING, " +
                "   `source_platform` STRING, " +
                "   `refund_status` INT, " +
                "   `meta` STRING, " +
                "   `hash_id` STRING, " +
                "   `trace_id` STRING, " +
                "   `appid` STRING, " +
                "   PRIMARY KEY (`id`) NOT ENFORCED" +
                ")with(" +
                "   'connector'='doris', " +
                "   'fenodes'='"+ OCPXConstant.OCPXDORISIP +"', " +
                "   'table.identifier'='"+ OCPXConstant.OCPXDORISDATABASE + "." + OCPXConstant.OCPXDWDPAYORDER +"', " +
                "   'username'='"+ OCPXConstant.OCPXDORISUSERNAME +"'," +
                "   'password'='"+ OCPXConstant.OCPXDORISPASSWORD +"'," +
                "   'sink.batch.size'='"+ OCPXConstant.OCPXDORISSINKBATCHSIZE +"', " +
                "   'sink.batch.interval'='"+ OCPXConstant.OCPXDORISSINKBATCHINTERVAL +"' " +
                ")";


        String insertTable = " insert into  dwd_ocpx_pay_order2 " +
                " select  " +
                " id, " +
                " payer_id, " +
                " product_id, " +
                " wk_account_id, " +
                " parent_activity_id, " +
                " src_prod_id, " +
                " org_id, " +
                " ks_aid, " +
                " product_name, " +
                " skin_id, " +
                " skin_name, " +
                " source_id, " +
                " source_name, " +
                " wxapp_id, " +
                " wxapp_nickname, " +
                " nickname, " +
                " headimgurl, " +
                " phone_num, " +
                " creative_id, " +
                " ad_plan_id, " +
                " ad_group_id, " + // MP消耗与订单关联必须做限制
                " gdt_group_id, " + //--广点通消耗表与订单表关联
                " bd_creative_id, " +
                " fee, " +
                " paid, " +
                " create_time, " +
                " pay_time, " +
                " subscribed, " +
                " tag_id, " +
                " tag_name , " +
                " source_platform, " +
                " refund_status, " +
                " meta, " +
                " hash_id, " +
                " trace_id, " +
                " appid " +
                " from( " +
                " select" +
                " t3.id,  " +
                " cast(null as int) as payer_id, " +
                " cast(null as int) as product_id, " +
                " cast(null as int) as wk_account_id, " +
                " cast(null as int) as parent_activity_id, " +
                " cast(null as int) as src_prod_id, " +
                " cast(null as int) as org_id, " +
                " cast(null as varchar) as ks_aid, " +
                " cast(null as varchar) as product_name, " +
                " cast(null as int) as skin_id, " +
                " cast(null as varchar) as skin_name, " +
                " cast(null as int) as source_id, " +
                " cast(null as varchar) as source_name, " +
                " cast(null as int) as wxapp_id, " +
                " cast(null as varchar) as wxapp_nickname, " +
                " cast(null as varchar) as nickname, " +
                " cast(null as varchar) as headimgurl, " +
                " cast(null as varchar) as phone_num, " +
                " cast(null as varchar) as creative_id, " +
                " cast(null as varchar) as ad_plan_id, " +
                " cast(null as varchar) as ad_group_id, " + // MP消耗与订单关联必须做限制
                " cast(null as varchar) as gdt_group_id, " + //--广点通消耗表与订单表关联
                " cast(null as varchar) as bd_creative_id, " +
                " cast(null as int) as fee, " +
                " cast(null as int) as paid, " +
                " cast(null as varchar) as create_time, " +
                " cast(null as varchar) as pay_time, " +
                " t2.subscribed as subscribed, " +
                " cast(null as int) as tag_id, " +
                " cast(null as varchar) as tag_name , " +
                " cast(null as varchar) as source_platform, " +
                " cast(null as int) as refund_status, " +
                " cast(null as varchar) as meta, " +
                " cast(null as varchar) as hash_id, " +
                " cast(null as varchar) as trace_id, " +
                " cast(null as varchar) as appid " +
                " from ocpx_account_account t1 " +
                " left join ods_ocpx_account_wxuser FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t2 " +
                "   on t1.parent_account_id = t2.account_id  " +
                " left join ocpx_pay_order FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t3 " +
                "   on t2.wxapp_id = t3.wxapp_id  " +
                "   and t1.id = t3.payer_id " +
                " left join ocpx_product_product FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t5 " +
                "   on cast(t3.product_id as int)= t5.id " +
                " where t5.src_prod_type = 'wk_act' " +
                " ) t";



        Map<String, String> res = new HashMap<>();
        res.put("source",sourceTable);
        res.put("sink",sinkTable);
        res.put("insert",insertTable);


        return res;
    }

}

